package com.rabbitmq.basicqos;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmListener;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.MessageProperties;
import com.rabbitmq.util.ConnectionUtils;
import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.util.Collections;
import java.util.SortedSet;
import java.util.TreeSet;
import java.util.concurrent.TimeoutException;

import static com.rabbitmq.common.Constant.NUM64;

/**
 * @className: QosProducer
 * @description: Qos生产者
 * @date: 2020/5/14
 * @author: cakin
 */
@Slf4j
public class QosProducer {
    public static void main( String[] args ) throws IOException, TimeoutException {
        Connection connection = ConnectionUtils.getConnection();
        Channel channel = connection.createChannel();

        channel.exchangeDeclare("exchange.confirm", "direct", true, false, null);
        channel.queueDeclare("queue.basicqos", true, false, false, null);
        channel.queueBind("queue.basicqos", "exchange.confirm", "routingKey2");

        final SortedSet<Long> confirmSet = Collections.synchronizedNavigableSet(new TreeSet<Long>());
        // 启动confirm模式
        channel.confirmSelect();

        // 开启监听
        channel.addConfirmListener(new ConfirmListener() {
            // 处理成功
            public void handleAck( long deliveryTag, boolean multiple ) throws IOException {
                log.info("消息发送成功: " + deliveryTag
                        + ", multiple : " + multiple);
                // 收到消息，从消息序号队列中删除deliveryTag
                if (multiple) {
                    confirmSet.headSet(deliveryTag - 1).clear();
                } else {
                    confirmSet.remove(deliveryTag);
                }
            }

            // 处理失败
            public void handleNack( long deliveryTag, boolean multiple ) throws IOException {
                if (multiple) {
                    confirmSet.headSet(deliveryTag - 1).clear();
                } else {
                    confirmSet.remove(deliveryTag);
                }
            }
        });

        // 循环发送消息
        for (int i = 1; i < 10000*NUM64; i++) {
            String msg = "我是confirm模式消息.异步[" + i + "]";
            long tag = channel.getNextPublishSeqNo();
            // 发送消息，将deliveryTag加入到消息序号队列
            channel.basicPublish("exchange.confirm", "routingKey2", MessageProperties.PERSISTENT_TEXT_PLAIN, msg.getBytes());
            log.info("tag:" + tag);
            confirmSet.add(tag);
        }
    }
}
